package com.jisf.driver.service.impl;

import com.alibaba.fastjson.JSONObject;
import com.jisf.driver.dao.DrivingSchoolLogDao;
import com.jisf.driver.entity.po.DrivingSchoolLog;
import com.jisf.driver.entity.vo.DrivingSchoolLogRequestVo;
import com.jisf.driver.entity.vo.DrivingSchoolLogResponseVo;
import com.jisf.driver.exception.SystemException;
import com.jisf.driver.service.DrivingSchoolLogService;
import org.apache.commons.lang.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.query.RangeQueryBuilder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.Page;
import org.springframework.data.domain.PageRequest;
import org.springframework.data.domain.Pageable;
import org.springframework.data.domain.Sort;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Optional;

/**
 * @Classname DrivingSchoolLogServiceImpl
 * @Description TODO
 * @Date 2022/5/24 19:07
 * @Created by jisf：【429627912@qq.com】
 */
@Service
public class DrivingSchoolLogServiceImpl implements DrivingSchoolLogService {

    @Autowired
    private DrivingSchoolLogDao drivingSchoolLogDao;

    /**
     * 存储日志信息
     * @param records
     * @param ack
     */
    @Override
    @KafkaListener(topics = {"${spring.kafka.topic}"}, groupId = "${spring.kafka.consumer.group-id}", containerFactory = "containerFactory")
    public void consumerLogInfo(List<ConsumerRecord<?, ?>> records, Acknowledgment ack) {
//        Object value = record.value();
//        System.out.println(value);
        records.forEach(record -> {
            Optional<?> kafkaMessage = Optional.ofNullable(record.value());
            kafkaMessage.ifPresent(u -> {
//                System.out.println("收到消息");
                String value = (String) u;
//                System.out.println(value);
                DrivingSchoolLog drivingSchoolLog = JSONObject.toJavaObject(JSONObject.parseObject(value), DrivingSchoolLog.class);
//                System.out.println("发送es：" + drivingSchoolLog);
                drivingSchoolLogDao.save(drivingSchoolLog);
//                System.out.println("成功发送es");
                ack.acknowledge();
//                System.out.println("ack确认机制");
            });
        });
    }

    /**
     * 查询日志信息
     * @param drivingSchoolLogRequestVo
     * @return
     */
    @Override
    public DrivingSchoolLogResponseVo selectLog(DrivingSchoolLogRequestVo drivingSchoolLogRequestVo) {
        try {
            Integer page = drivingSchoolLogRequestVo.getPage();
            Integer size = drivingSchoolLogRequestVo.getSize();
            String name = drivingSchoolLogRequestVo.getName();
            String msg = drivingSchoolLogRequestVo.getMsg();
            String date = drivingSchoolLogRequestVo.getDate();
            Boolean status = drivingSchoolLogRequestVo.getStatus();
            // 排序
            Sort sort = Sort.by(Sort.Direction.DESC, "date");
            // 分页
            Pageable pageable = PageRequest.of(page-1, size, sort);
            // 组合查询bool
            BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
            if (StringUtils.isNotBlank(name)) {
                boolQueryBuilder.must(QueryBuilders.matchQuery("name", name));
            }
            if (StringUtils.isNotBlank(msg)) {
                boolQueryBuilder.must(QueryBuilders.matchQuery("msg", msg));
            }
            if (StringUtils.isNotBlank(date)) {
                String[] dateArray = date.split(" ");
                RangeQueryBuilder rangeQuery = QueryBuilders.rangeQuery("date");
                rangeQuery.gte(dateArray[0]).lte(dateArray[1]);
                boolQueryBuilder.must(rangeQuery);
            }
            if (status != null) {
                boolQueryBuilder.must(QueryBuilders.termQuery("status", status));
            }

            Page<DrivingSchoolLog> logPage = drivingSchoolLogDao.search(boolQueryBuilder, pageable);
            List<DrivingSchoolLog> content = logPage.getContent();

            long totalElements = logPage.getTotalElements();
            return new DrivingSchoolLogResponseVo(totalElements, content);
        }catch (Exception e){
            throw new SystemException(e);
        }

    }
}
